package xin.alum.aim.rabbitmq;

import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import xin.alum.aim.constant.AimConstant;
import xin.alum.aim.constant.Constants;
import xin.alum.aim.groups.SessionGroups;
import xin.alum.aim.groups.Sessions;

import java.util.UUID;

/**
 * @author alum
 */
@Configuration
@RequiredArgsConstructor
@ConditionalOnProperty(prefix = AimConstant.CLUSTER_PROPERTIES_PREFIX, name = AimConstant.CLUSTER_PROPERTIES_MODE, havingValue = "rabbitmq")
class RabbitMQConfig {

    protected final InternalLogger logger = InternalLoggerFactory.getInstance(this.getClass());

    private final ConnectionFactory connectionFactory;

    /**
     * 针对消费者配置:
     * <p>
     * 1. 设置交换机类型。
     * <p>
     * 2. 将队列绑定到交换机。
     * <p>
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念。
     * <p>
     * HeadersExchange ：通过添加属性key-value匹配。
     * <p>
     * DirectExchange:按照routingkey分发到指定队列。
     * <p>
     * TopicExchange:多关键字匹配。
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("xin.alum.aim.cluster.fanout", true, true);
    }

    @Bean
    public Queue reply() {
        return new Queue(Constants.EVENT_MESSAGE_INNER_QUEUE, true);// 队列持久化
    }

    @Bean
    public Queue message() {
        return new Queue(Constants.PUSH_MESSAGE_INNER_QUEUE, true);// 队列持久化
    }

    @Bean
    public Queue kick() {
        return new Queue(Constants.BIND_MESSAGE_INNER_QUEUE, true);// 队列持久化
    }

    @Bean
    public Binding bindReply() {
        return BindingBuilder.bind(reply()).to(fanoutExchange());
    }

    @Bean
    public Binding bindMessage() {
        return BindingBuilder.bind(message()).to(fanoutExchange());
    }

    @Bean
    public Binding bindKick() {
        return BindingBuilder.bind(kick()).to(fanoutExchange());
    }

    @Bean
    public RabbitMessageListener rabbitMessageListener(SessionGroups groups, Sessions sessions) {
        return new RabbitMessageListener(groups,sessions);
    }


    // RabbitTemplate，即消息模板，我们在与SpringAMQP整合的时候进行发送消息的关键类。
    // 该类提供了丰富的发送消息方法,包括可靠性投递消息方法,回调监听消息接口ConfirmCallback,
    // 返回值确认ReturnCallBack等等。同样我们需要进行注入到Spring容器中,然后直接使用。
    // 将RabbitTemplate加入到Spring容器中
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        //confirm 监听，当消息成功发到交换机 ack = true，没有发送到交换机 ack = false
        //correlationData 可在发送时指定消息唯一 id
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                //记录日志、发送邮件通知、落库定时任务扫描重发
                logger.error("交换机数据丢失：{}", correlationData);
            }
        });

        //当消息成功发送到交换机没有路由到队列触发此监听
        rabbitTemplate.setReturnsCallback(returned -> {
            //记录日志、发送邮件通知、落库定时任务扫描重发
            logger.error("队列监听数据丢失：{}", returned.getMessage());
        });
        return rabbitTemplate;
    }

    /**
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(RabbitMessageListener rabbitMessageListener) {
        final int MAX_CONCURRENT_CONSUMERS = 50;
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        // 监听的队列,可变参数,可以添加多个队列。
        container.setQueues(reply(), kick(), message());
        // 设置消费者个数。
        container.setConcurrentConsumers(MAX_CONCURRENT_CONSUMERS / 5);
        // 设置最大消费者个数。
        container.setMaxConcurrentConsumers(MAX_CONCURRENT_CONSUMERS);
        // 设置默认是否重回队列。
        container.setDefaultRequeueRejected(false);
        // 设置签收模式,自动签收。
        container.setAcknowledgeMode(AcknowledgeMode.AUTO);
        // 设置是否外露
        container.setExposeListenerChannel(true);
        // 设置标签策略。
        container.setConsumerTagStrategy(queue -> queue.concat("_").concat(UUID.randomUUID().toString()));
        //container.setPrefetchCount(300);
        // 监听消息
        container.setMessageListener(rabbitMessageListener);
        return container;
    }

    @Bean
    public RabbitMQPusher rabbitMQPusher(RabbitTemplate rabbitTemplate) {
        return new RabbitMQPusher(rabbitTemplate);
    }
}
